---
title: Actor模型
description: 深入理解CActor框架的核心 - Actor模型的设计原理和实现机制
category: 核心概念
order: 1
tags: [Actor模型, 并发, 消息传递, 容错]
author: CActor团队
publishedAt: 2024-01-15
updatedAt: 2024-01-15
featured: true
difficulty: intermediate
---

# Actor模型

Actor模型是CActor框架的核心设计理念，它提供了一种优雅的方式来处理并发、分布式和容错系统。

## 什么是Actor模型

Actor模型是一种并发计算的数学模型，由Carl Hewitt在1973年提出。在这个模型中，Actor是计算的基本单元，它们通过异步消息传递进行通信。

### 核心特性

- **封装性**: 每个Actor都有私有状态，外部无法直接访问
- **异步通信**: Actor之间通过消息异步通信，避免共享状态
- **位置透明**: Actor可以在本地或远程运行，调用方式相同
- **容错性**: Actor失败不会影响其他Actor的运行

## Actor的基本概念

### Actor的组成

每个Actor包含以下核心组件：

```cangjie
class MyActor : Actor {
    // 私有状态
    var counter: Int32 = 0
    var data: Map<String, String> = Map()

    // 消息处理函数
    override func receive(message: Message): Unit {
        match (message) {
            case IncrementMessage() => {
                counter += 1
                println("Counter: ${counter}")
            }
            case GetCounterMessage() => {
                sender.send(CounterResponse(counter))
            }
            case SetDataMessage(key, value) => {
                data[key] = value
            }
        }
    }

    // 生命周期钩子
    override func preStart(): Unit {
        println("Actor starting...")
    }

    override func postStop(): Unit {
        println("Actor stopped.")
    }
}
```

### Actor的生命周期

```mermaid
graph TD
    A[创建] --> B[preStart]
    B --> C[运行中]
    C --> D[接收消息]
    D --> E[处理消息]
    E --> C
    C --> F[停止]
    F --> G[postStop]
    G --> H[销毁]

    C --> I[异常]
    I --> J[重启]
    J --> K[preRestart]
    K --> L[postRestart]
    L --> C
```

## 消息传递机制

### 消息定义

在CActor中，消息是不可变的数据结构：

```cangjie
// 简单消息
struct PingMessage {}

struct PongMessage {}

// 带参数的消息
struct TransferMessage {
    let from: String
    let to: String
    let amount: Float64
    let transactionId: String
}

// 响应消息
struct TransferResponse {
    let success: Bool
    let transactionId: String
    let message: String
}
```

### 发送消息

```cangjie
// 异步发送（Fire-and-forget）
actor.send(PingMessage())

// 请求-响应模式
let future = actor.ask<TransferResponse>(
    TransferMessage("Alice", "Bob", 100.0, "tx-001")
)

// 等待响应
match (future.await(timeout: 5.seconds)) {
    case Some(response) => {
        if (response.success) {
            println("Transfer successful: ${response.message}")
        } else {
            println("Transfer failed: ${response.message}")
        }
    }
    case None => {
        println("Request timeout")
    }
}
```

### 消息路由

```cangjie
class RouterActor : Actor {
    let workers: Array<ActorRef> = []
    var currentIndex: Int32 = 0

    override func receive(message: Message): Unit {
        match (message) {
            case WorkMessage(data) => {
                // 轮询路由
                let worker = workers[currentIndex]
                worker.send(ProcessMessage(data))
                currentIndex = (currentIndex + 1) % workers.size()
            }
            case AddWorkerMessage(worker) => {
                workers.append(worker)
            }
        }
    }
}
```

## Actor层次结构

### 父子关系

CActor采用层次化的Actor系统：

```cangjie
class SupervisorActor : Actor {
    var workers: Array<ActorRef> = []

    override func preStart(): Unit {
        // 创建子Actor
        for (i in 0..10) {
            let worker = context.actorOf<WorkerActor>("worker-${i}")
            workers.append(worker)
        }
    }

    override func receive(message: Message): Unit {
        match (message) {
            case DistributeWorkMessage(tasks) => {
                for ((task, worker) in tasks.zip(workers)) {
                    worker.send(ProcessTaskMessage(task))
                }
            }
        }
    }

    // 监督策略
    override func supervisorStrategy(): SupervisorStrategy {
        return OneForOneStrategy(
            maxRetries: 3,
            withinTimeRange: 1.minute,
            decider: { exception =>
                match (exception) {
                    case _: IllegalArgumentException => Restart
                    case _: IOException => Resume
                    case _ => Stop
                }
            }
        )
    }
}
```

### 监督策略

```cangjie
// 一对一策略：只影响失败的子Actor
OneForOneStrategy(
    maxRetries: 3,
    withinTimeRange: 1.minute,
    decider: { exception => Restart }
)

// 一对多策略：影响所有子Actor
AllForOneStrategy(
    maxRetries: 1,
    withinTimeRange: 30.seconds,
    decider: { exception => Restart }
)
```

## 并发和性能

### 无锁并发

Actor模型通过消息传递避免了共享状态，实现了无锁并发：

```cangjie
class CounterActor : Actor {
    var count: Int64 = 0

    override func receive(message: Message): Unit {
        match (message) {
            case IncrementMessage() => {
                // 无需加锁，Actor内部是单线程的
                count += 1
            }
            case GetCountMessage() => {
                sender.send(CountResponse(count))
            }
        }
    }
}
```

### 背压控制

```cangjie
class BackpressureActor : Actor {
    let maxQueueSize: Int32 = 1000
    var currentQueueSize: Int32 = 0

    override func receive(message: Message): Unit {
        match (message) {
            case WorkMessage(data) => {
                if (currentQueueSize >= maxQueueSize) {
                    sender.send(BackpressureMessage("Queue full, try later"))
                    return
                }

                currentQueueSize += 1
                processWork(data)
                currentQueueSize -= 1
            }
        }
    }

    func processWork(data: String): Unit {
        // 处理工作
        Thread.sleep(100) // 模拟处理时间
    }
}
```

## 容错机制

### 异常处理

```cangjie
class ResilientActor : Actor {
    override func receive(message: Message): Unit {
        try {
            match (message) {
                case ProcessMessage(data) => {
                    processData(data)
                }
            }
        } catch (e: Exception) {
            // 记录错误
            logger.error("Processing failed: ${e.message}")

            // 发送错误响应
            sender.send(ErrorResponse(e.message))

            // 抛出异常触发监督策略
            throw e
        }
    }

    func processData(data: String): Unit {
        if (data.isEmpty()) {
            throw IllegalArgumentException("Data cannot be empty")
        }
        // 处理数据
    }
}
```

### 断路器模式

```cangjie
class CircuitBreakerActor : Actor {
    let circuitBreaker = CircuitBreaker(
        maxFailures: 5,
        callTimeout: 10.seconds,
        resetTimeout: 1.minute
    )

    override func receive(message: Message): Unit {
        match (message) {
            case ExternalCallMessage(request) => {
                circuitBreaker.withCircuitBreaker {
                    makeExternalCall(request)
                }.onComplete { result =>
                    match (result) {
                        case Success(response) => {
                            sender.send(SuccessResponse(response))
                        }
                        case Failure(exception) => {
                            sender.send(ErrorResponse(exception.message))
                        }
                    }
                }
            }
        }
    }
}
```

## 分布式Actor

### 远程Actor

```cangjie
// 创建远程Actor引用
let remoteActor = system.actorSelection(
    "cactor://RemoteSystem@192.168.1.100:8080/user/remoteActor"
)

// 发送消息到远程Actor
remoteActor.send(RemoteMessage("Hello from local"))
```

### 集群配置

```hocon
cactor {
  cluster {
    seed-nodes = [
      "cactor://ClusterSystem@192.168.1.100:8080",
      "cactor://ClusterSystem@192.168.1.101:8080"
    ]

    roles = ["frontend", "backend"]

    auto-down-unreachable-after = 10s
  }

  remote {
    enabled-transports = ["cactor.remote.netty.tcp"]
    netty.tcp {
      hostname = "192.168.1.100"
      port = 8080
    }
  }
}
```

## 最佳实践

### 1. 保持Actor轻量

```cangjie
// 好的做法：专注单一职责
class UserValidatorActor : Actor {
    override func receive(message: Message): Unit {
        match (message) {
            case ValidateUserMessage(user) => {
                let isValid = validateUser(user)
                sender.send(ValidationResult(isValid))
            }
        }
    }
}

// 避免：承担过多职责
class MonolithicActor : Actor {
    // 避免在一个Actor中处理太多不相关的功能
}
```

### 2. 使用不可变消息

```cangjie
// 好的做法：不可变消息
struct UserCreatedEvent {
    let userId: String
    let username: String
    let timestamp: DateTime
}

// 避免：可变消息
class MutableMessage {
    var data: String // 避免可变字段
}
```

### 3. 合理设计消息协议

```cangjie
// 使用密封类定义消息协议
sealed class BankAccountMessage

struct DepositMessage(amount: Float64) : BankAccountMessage
struct WithdrawMessage(amount: Float64) : BankAccountMessage
struct GetBalanceMessage() : BankAccountMessage
struct CloseAccountMessage() : BankAccountMessage
```

## 性能考虑

### 消息吞吐量

- **单个Actor**: 每秒可处理数百万条消息
- **系统总体**: 支持数十亿条消息/秒
- **延迟**: 微秒级消息传递延迟

### 内存使用

- **Actor实例**: 每个Actor约占用400字节
- **消息队列**: 动态调整，支持背压控制
- **垃圾回收**: 优化的内存管理，减少GC压力

## 总结

Actor模型为CActor框架提供了强大的并发和分布式计算能力：

- **简化并发编程**: 无需考虑锁和共享状态
- **天然容错**: 隔离失败，快速恢复
- **水平扩展**: 轻松扩展到多机集群
- **位置透明**: 本地和远程调用统一

通过理解和掌握Actor模型，你可以构建出高性能、高可用的分布式系统。

## 下一步

- [学习消息传递模式](/docs/core-concepts/messaging)
- [了解Actor生命周期](/docs/core-concepts/lifecycle)
- [探索监督策略](/docs/core-concepts/supervision)
- [查看性能优化指南](/docs/best-practices/performance)
